import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.json4s.JsonDSL._
import org.json4s.jackson.JsonMethods._
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}
object cleanedsourceds {
   //1.创建Spark环境配置对象
   val conf = new SparkConf().setAppName("cleanedsourceds").setMaster("local")
   //2.创建SparkSession对象
   val spark:SparkSession = SparkSession.builder().config(conf).getOrCreate()

   import spark.implicits._
   var commentData: DataFrame = spark.read.format("csv")
     .option("header", true)
     .option("multiLine", true)
     .load("file:///D:\\IDEA_Projects\\sparkexample_v1\\src\\input\\tnb.csv")

//   var commentData: DataFrame = spark.read.format("csv")
//     .option("header", true)
//     .option("multiLine", true)
//     .load("file:///musicdata/comments.csv")
   //3.注册临时表
   commentData.createOrReplaceTempView("tnbsource")
   //4.查询操作
   val sqlresult : DataFrame=
      spark.sql("select substring(time,12,2) as commenthour,count(*) as commenttotal " +
        "   from tnbsource " +
        "   group by substring(time,12,2) " +
        "   order by substring(time,12,2) desc")

   sqlresult.show()
   //5.将分析结果保存到数据表中
   //       eg:存入oa数据库中
   sqlresult.write
     .format("jdbc")
     .option("url","jdbc:mysql://localhost/health_monitoring")
     .option("user","root")
     .option("password","root")
     .option("dbtable","tnb_cleaned")
     .mode(SaveMode.Append)
     .save()

}
